package org.robby.hbase.cdr.filter;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FamilyFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.QualifierFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.robby.hbase.cdr.HbaseCdrIF;

public class FamilyAndQualifierFilterTest {

	private static HTablePool pool;
	public static HbaseCdrIF hbsase = null;
	private static String tab_cdr_name = "tab_cdr2";

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		HBaseConfiguration config = new HBaseConfiguration();
		config.set("hbase.zookeeper.quorum", "hadoop.main");// 连接到远程服务上
		pool = new HTablePool(config, 1000);

		Scan s = new Scan();
		SingleColumnValueFilter filter1 = new SingleColumnValueFilter(Bytes.toBytes("data"),// 列族
				Bytes.toBytes("oareacode"),// qualifier
				CompareFilter.CompareOp.NOT_EQUAL,// 比较
				new SubstringComparator("010")// 比较器
		);
		// 匹配row(key)的过滤器
		Filter filter2 = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(".*_.*_2012080."));
		// 匹配列族的过滤器
		Filter filter3 = new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("data")));
		// 匹配列的过滤器
		Filter filter4 = new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("oaddr")));
		// 匹配值的过滤器
		Filter filter5 = new ValueFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator("010"));
		// 匹配所有主叫码为13900000161记录。
		// PrefixFilter与Scan的setStartRow(key)和setStopRow(key)功能相同
		Filter filter6 = new PrefixFilter(Bytes.toBytes("13900000161"));
		//分页过滤器
		Filter filter7 = new PageFilter(10);
		//scan的row key规则：start_row <= scan < stop_row
		//把第一页的最后一个row key 取出并加1之后，就可以从第排除第一个
		s.setStartRow(Bytes.add("13900000000_13900007704_20120822120714".getBytes(), Bytes.toBytes(1)));

		FilterList fl = new FilterList(FilterList.Operator.MUST_PASS_ONE);
		fl.addFilter(filter1);
		fl.addFilter(filter2);

		s.setFilter(filter7);

		HTableInterface tab_cdr2 = pool.getTable(tab_cdr_name);
		ResultScanner rs = tab_cdr2.getScanner(s);

		for (Result r : rs) {
			String out = Bytes.toString(r.getRow()) + "\t";
			byte[] data = r.getValue(Bytes.toBytes("data"), Bytes.toBytes("oaddr"));
			out += Bytes.toString(data) + "\t";
			data = r.getValue(Bytes.toBytes("data"), Bytes.toBytes("oareacode"));
			out += Bytes.toString(data) + "\t";
			data = r.getValue(Bytes.toBytes("data"), Bytes.toBytes("daddr"));
			out += Bytes.toString(data) + "\t";
			data = r.getValue(Bytes.toBytes("data"), Bytes.toBytes("dareacode"));
			out += Bytes.toString(data) + "\t";
			data = r.getValue(Bytes.toBytes("data"), Bytes.toBytes("ts"));
			out += Bytes.toString(data) + "\t";
			data = r.getValue(Bytes.toBytes("data"), Bytes.toBytes("type"));
			out += Bytes.toString(data) + "\t";
			System.out.println(out);
		}

	}

}
